/*
 * Transfer.cpp
 *
 *  Created on: Jun 7, 2011
 *      Author: Mario Hock
 */

#include "Transport.h"

#include "messages/ConnectMessage.h"
#include "messages/ConnectConfirmMessage.h"

#include <boost/foreach.hpp>
#include "UI_multiplexer.h"

// random
#include <stdlib.h>
#include <time.h>


#define for_each BOOST_FOREACH


/** constructor / destructor **/
Transport::Transport(ariba::AribaModule* ariba, ariba::Node* node, Userdatabase* users) :
    ariba(ariba), node(node), users(users), /*mcpoReady(false),*/ joinFailedFlag(false)
{
    serviceID = ariba::utility::ServiceID( 555 );
    node->bind(this, serviceID);

    //TODO, größe 10 kann ja nicht sinn der sache sein. sollte abhängig von dem message types sein
    for (int i=0; i<10; i++)
        messageHandler[i] = NULL;
    
    this->ui = &UI_multiplexer::instance(); //FIXME always use singleton
    
    /* initialize random seed: */
    srand ( time(NULL) );
}

Transport::~Transport()
{
}


/* save status */
void Transport::save_arq_queue(const path& message_dir)
{
    arq_queue.save_to_disk(message_dir);
}

/* load status */
void Transport::load_arq_queue(const path & message_dir)
{
    arq_queue.load_from_disk(message_dir);
}



/** SENDING **/
/* send over connection */
// private function (raw version, doesn't check connection)
void Transport::_send_to(const ariba::DataMessage& msg, TransportMessage::MESSAGE_TYPE type, uint8_t subtype, UserPtr user)
{
    // Encapsulate in TransportMessage
    TransportMessage tmsg(type, subtype);
    tmsg.encapsulate(msg);

    // send message
    node->sendMessage(tmsg, user->link);
}

// public function (checks connection)
bool Transport::send_to(const ariba::DataMessage& msg, TransportMessage::MESSAGE_TYPE type, uint8_t subtype, UserPtr user)
{
    // error if not connected
    if ( ! user->is_connected() )
    {
        UI.write_debug("send_to: Can not send message to user: '" + user->get_name() + "'");
        return false;
    }

    // call raw function
    _send_to(msg, type, subtype, user);
    
    return true;
}


// TODO
//  --> receipt
//  --> duplicates..?
//
void Transport::send_eventually(const ariba::DataMessage& msg, TransportMessage::MESSAGE_TYPE type, uint8_t subtype, UserPtr user, string receipt)
{
    // generate sequence number
    omnibus_seq_num seq = arq_queue.get_next_seqnum(user);
    
    // Encapsulate in TransportMessage
    TMsgPtr tmsg(new TransportMessage(type, subtype));
    tmsg->set_seq_num(seq);
    tmsg->request_ack();
    tmsg->encapsulate(msg);
    
    // store message
    arq_queue.push_message(user, seq, tmsg);
    
    
    // BRANCH: user connected --> send message
    if ( user->is_connected() )
    {
        // TODO erst »arq_on_connection_established(user)«, falls noch alte messages in der queue..?
        
        node->sendMessage(*tmsg, user->link);
    }
    // BRANCH: user not connected --> try to connect
    else
    {
        establish_connection(user);
    }
}


void Transport::arq_on_connection_established(UserPtr user)
{
    // get waiting messages from arq_queue
    vector<TMsgPtr> messages;
    arq_queue.get_waiting_messages(messages, user);
    
    // send waiting messages
    for_each( TMsgPtr m, messages )
    {
        node->sendMessage(*m, user->link);
        
//        // XXX
//        ostringstream out;
//        out << "---- SENDING: ";
//        out << m->get_seq_num();
//        UI.write_debug(out.str());
    }
}


/* send unconnected */
//void Transport::send_direct(const ariba::DataMessage& msg, TransportMessage::MESSAGE_TYPE type, uint8_t subtype, UserPtr user)
//{
//    // Encapsulate in TransportMessage
//    TransportMessage tmsg(type, subtype);
//    tmsg.encapsulate(msg);
//
//    // Send message to "addr"
//    node->sendMessage( tmsg, user->get_id(), serviceID );
//}


/* creates an ariba link to a user, and sets status CONNECTED */
void Transport::establish_connection(UserPtr user)
{
    // XXX testing
//    if ( user->link.valid() )
//        node->dropLink(user->link);

    // set connection status to "pending"
    user->set_connected(User::CONNECTION_PENDING);

    // establish ariba link -> callback on success or failure
    ariba::utility::LinkID lnk = node->establishLink(user->get_id(), serviceID);

    ostringstream out;
    out << "[USER_UP] connecting to '" << user->get_name() << "' (LinkID: " << lnk << ") ...";
    UI.write_to_user(out.str());
}

/* gracefully closes an connection */
void Transport::close_connection(UserPtr user)
{
    // TODO nur wenn connected? cornercases?
    // already not connected
//    if ( ! user->is_connected() )
//        return;

//    user->set_connected(User::DISCONNECT_PENDING);
    UI.write_to_user("[USER_UP] disconnecting '" + user->get_name() + "' ...");
    notify_conn_stat(user, false);

    close_connection_raw(user);
}

/* sends a disconnect message, awaits link down callback */
void Transport::close_connection_raw(UserPtr user)
{
    ChatMessage m(""); // TODO hier würd ich gern gar keine message mitschicken.. segfault..?
    send_to(m, TransportMessage::CONNECTION, Transport::DISCONNECT, user);
    user->set_connected(User::DISCONNECT_PENDING);
}

/* calls close_connection for every connected user */
void Transport::close_all_connections()
{
    // TODO wenn wir die offenen Verbundungen selber verwalten, müssen wir hier nicht auf die gesamte userdb zugreifen..
    Userdatabase::UserMap db = users->get_whole_database();

    // for all users
    for ( Userdatabase::UserMap::iterator it = db.begin() ; it != db.end(); it++ )
    {
        UserPtr u = it->second;
        if ( u->is_connected() )
        {
            // hier ist keine disconnect_notification, weil ein (ggf. effizienteres) disconnect_all vorher erwarted wird
            close_connection_raw(u);
        }
    }

//    UI.write_to_user("[USER_UP] all connections closing.");  // FIXME Absturz wenn client disconnected
    cout << "[USER_UP] all connections closing." << endl;
}

/* notifies upper layer about connect/disconnect (true means "connect") */
void Transport::notify_conn_stat(UserPtr user, bool connected)
{
    /* notify upper layer */
	vector<ConnStatInterface*>::iterator it;

	for ( it=conn_stat_notification.begin(); it < conn_stat_notification.end(); it++ )
	{
		(*it)->got_connection_update(user, connected);
	}
}


/** RECEIVING **/
void Transport::handleMessage(const ariba::DataMessage& msg, const ariba::NodeID& sender_id, const ariba::LinkID on_link)
{
  TransportMessage* transmsg = msg.getMessage()->convert<TransportMessage> ();
  TransportMessage::MESSAGE_TYPE type = transmsg->getType();
  uint8_t subtype = transmsg->getSubtype();
  UserPtr user = users->getUser(sender_id);

  
  // Connection messages
  if ( type == TransportMessage::CONNECTION )
  {
      // OK: known user
      if ( user )
      {
          handle_connect(transmsg, type, subtype, user);
      }
      // ERR: unknown user
      else
      {
          UI.write_info("Dropping incoming connection from unknown user: '" + sender_id.getAddress() + "'");
      }
      
      delete transmsg;
      return;
  }

  

  /* some error cases */
  // message from unknown user. (could lead to crash when referencing to null-pointer, so we check first)
  if ( !user )
  {
      UI.write_warn("WARNING: Message from unknown user: '" + sender_id.getAddress() + "'");
      delete transmsg;
      return;
  }

  // message from non-connected user -- this should never happen
  if ( ! user->is_connected() )
  {
      UI.write_warn("WARNING: Message non-connected user: '" + sender_id.getAddress() + "' -- This should never happen! Message dropped.");
      delete transmsg;
      return;
  }

  
  
  /* message flags */
  // ack requested
  if ( transmsg->ack_requested() )
  {
//      // XXX debug output
//      ostringstream out;
//      out << "||| GOT MESSAGE with SEQ_NUM: " << transmsg->get_seq_num();
//      UI.write_debug(out.str());

      
      // send ack
      TransportMessage ack_msg(TransportMessage::ACK, type);
      ack_msg.set_seq_num(transmsg->get_seq_num());

      // send message
      node->sendMessage(ack_msg, user->link);
  }
  
  
//  // XXX Testing.. If we haven't got a Link Up event for this link yet, create one of our own..
//  // be careful, it could be an old link. only update if we wern't connected yet
//  if ( user->link == ariba::LinkID::UNSPECIFIED )
//  {
//      onLinkUp(on_link, sender_id);
//  }
  // XXX ----> macht nur ärger!

  
  
  // BRANCH: is ack --> handle here
  if ( type == TransportMessage::ACK )
  {
      omnibus_seq_num seq = transmsg->get_seq_num();
      
//      // XXX debug output
//      ostringstream out;
//      out << "||| GOT ACK for: " << seq;
//      UI.write_debug(out.str());
      
      arq_queue.remove_message(user, seq);
      // TODO receipt
      
      delete transmsg;
  }
  
  // BRANCH: message handler
  else if ( messageHandler[type] )
  {
      messageHandler[type]->handleMessage(transmsg, type, subtype, user);
  }

  // BRANCH: handle message here
  else
  {
    switch (type)
    {
        // ---> s.o.
//        // DISCONNECT
//        case TransportMessage::CONNECTION:
//        {
//            handle_connect(subtype, user);
//            break;
//        }
        
        default:
        {
            if ( ui )
              UI.write_to_user("ERROR: unknown message type: " + transmsg->getType());

            break;
        }
    }

    delete transmsg;
  }
}

// handle CONNECTION-message
void Transport::handle_connect(TransportMessage* msg, TransportMessage::MESSAGE_TYPE type, uint8_t subtype, UserPtr user)
{
    switch ( subtype )
    {
        /* connection request from another user */
        case Transport::CONNECT:
        {
            ConnectMessage* m = msg->decapsulate<ConnectMessage>();
            
            // OK: version compatible
            if ( m->is_fine() )
            {
                UI.write_info("Omnibus version of user '" + user->get_name() + "' compatible. ---> Connection accepted.");
                
                /* send connect-confirm message */
                ConnectConfirmMessage ans(true, m->get_nonce());
                
                // Encapsulate in TransportMessage
                TransportMessage tmsg(TransportMessage::CONNECTION, Transport::CONFIRM_CONNECTION);
                tmsg.encapsulate(&ans);

                // send message
                node->sendMessage(tmsg, user->link);

                
                // update status
                switch ( user->get_connection_status() )
                {
                    case User::NOT_CONNECTED:
                    case User::ERROR_NOT_CONNECTED:
                    case User::AUTHENTICATION_PENDING:
                    {
                        user->set_connected(User::RECEIVE_ESTABLISHED);
                        
                        break;
                    }
                    // ----> CONNECTED !!
                    case User::SEND_ESTABLISHED:
                    {
                        user->set_connected(User::CONNECTED);
                        UI.write_to_user("[USER_UP] '" + user->get_name() + "' connected!");
                        
                        arq_on_connection_established(user);
                        notify_conn_stat(user, true);
                        
                        break;
                    }
                    case User::CONNECTION_ERROR:
                    {
                        UI.write_warn("Would accept connection with '" + user->get_name() + "', but we're already in error state!");
                        
                        break;
                    }
                    
                    // should not happen! 
                    default:
                    {
                        ostringstream out;
                        out << "ERROR: Unexpected connection state (";
                        out << (int) user->get_connection_status();
                        out << ") while getting 'CONNECT' message.";
                        UI.write_warn(out.str());

                        user->set_connected(User::CONNECTION_ERROR);
                        node->dropLink(user->link);
                        
                        break;
                    }
                }
            }
            // ERR: version incompatible
            else
            {
                user->set_connected(User::CONNECTION_ERROR);
                UI.write_to_user("[USER_UP]");
                UI.write_warn("Connection error '" + user->get_name() + "' (wrong Omnibus version)!");
                
                ConnectConfirmMessage ans(false, m->get_nonce());
                _send_to(&ans, TransportMessage::CONNECTION, DENY_CONNECTION, user);
                
                node->dropLink(user->link);
            }
            
            delete m;
            break;
        }
        
        /* answer from other user: our connection request is accepted */
        case Transport::CONFIRM_CONNECTION:
        {
//            ConnectConfirmMessage* m = msg->decapsulate<ConnectConfirmMessage>();
            
            // update status
            switch ( user->get_connection_status() )
            {
                case User::NOT_CONNECTED:
                case User::ERROR_NOT_CONNECTED:
                case User::AUTHENTICATION_PENDING:
                {
                    user->set_connected(User::SEND_ESTABLISHED);
                    
                    break;
                }
                // ----> CONNECTED !!
                case User::RECEIVE_ESTABLISHED:
                {
                    user->set_connected(User::CONNECTED);
                    UI.write_to_user("[USER_UP] '" + user->get_name() + "' connected!");
                    
                    arq_on_connection_established(user);
                    notify_conn_stat(user, true);
                    
                    break;
                }
                case User::CONNECTION_ERROR:
                {
                    UI.write_warn("Would accept connection with '" + user->get_name() + "', but we're already in error state!");
                    
                    break;
                }
                
                // should not happen! 
                default:
                {
                    UI.write_warn("ERROR: Unexpected connection state while getting 'CONNECT' message.");

                    user->set_connected(User::CONNECTION_ERROR);
                    node->dropLink(user->link);
                    
                    break;
                }
            }
            
//            delete m;
            break;
        }
        
        /* answer from other user: our connection request is DENIED */
        case Transport::DENY_CONNECTION:
        {
            user->set_connected(User::CONNECTION_ERROR);
            UI.write_to_user("[USER_UP] Connection error '" + user->get_name() + "' (wrong Omnibus version)!");
                        
            node->dropLink(user->link);
            
            break;
        }
        
        case Transport::AUTHENTICATION_ERROR:
        {
            user->set_connected(User::CONNECTION_ERROR);
            UI.write_to_user("[USER_UP] Authentication error (" + user->get_name() + ")");
            UI.write_warn("Authentication error (" + user->get_name() + ")");
                        
            node->dropLink(user->link);
            
            break;
        }
        
        // disconnect
        case Transport::DISCONNECT:
        {
            // BRANCH: user in database
            if ( user )
            {
                user->set_connected(User::DISCONNECT_PENDING);
                UI.write_to_user("[USER_UP] disconnecting '" + user->get_name() + "' ...");
                notify_conn_stat(user, false);

                // drop ariba link, awaits callback
                node->dropLink(user->link);
            }

            break;
        }
    }
}


void Transport::register_message_handler(MessageHandlerInterface* handler, TransportMessage::MESSAGE_TYPE type)
{
//    // resize if necessary
//    if ( messageHandler.capacity() < type )
//        messageHandler.resize(type);

    messageHandler[type] = handler;
}

void Transport::register_conn_stat_notification(ConnStatInterface* handler)
{
    conn_stat_notification.push_back(handler);
}


// shows if there are neighbors in the overlay
string Transport::overlay_info()
{
    if ( node->getNeighborNodes().size() > 0 )
    {
        return "[ONLINE] true";
    }
    else
    {
        return "[ONLINE] false";
    }

//        ostringstream out;
//        out << endl;
//        for_each ( ariba::NodeID n, node->getNeighborNodes() )
//        {
//            out << "Neighbor: " << n.getAddress() << endl;
//        }
//
//    return out.str();
}



/**
 * Communication Listener (Service 555)
 */

/* INCOMING DATA - Service 555 */
void Transport::onMessage(const ariba::DataMessage& msg, const ariba::NodeID& remote, const ariba::LinkID& lnk)
{
    handleMessage(msg, remote, lnk);
}

void Transport::onLinkUp(const ariba::LinkID& lnk, const ariba::NodeID& lscrew)
{
    if ( ! users )
        return;
    
    
    UserPtr user = users->getUser(lscrew);

    // BRANCH: link from known user -> set CONNECTED
    if ( user )
    {
        if ( user->link != ariba::LinkID::UNSPECIFIED )
        {
            ;//TODO den alten link wegschmeißen?
        }

        user->link = lnk;
        user->set_connected(User::AUTHENTICATION_PENDING);
        UI.write_info("Link with '" + user->get_name() + "' established. Sending greetings ...");
        
        
        /* send connect message */
        ConnectMessage msg(rand());
        
        // Encapsulate in TransportMessage
        TransportMessage tmsg(TransportMessage::CONNECTION, Transport::CONNECT);
        tmsg.encapsulate(&msg);

        // send message
        node->sendMessage(tmsg, lnk);
    }
    else
    // BRANCH: link from unknown user -> DENY
    {
        // create temp user (deprecated)
////        UserPtr user = users->createTempUser(lscrew);
//        user = users->createTempUser(lscrew);
//        user->link = lnk;
//        user->set_connected(User::AUTHENTICATION_PENDING);
//        UI.write_to_user("[USER_UP] connection from TEMPUSER (" + user->get_id().getAddress() + ")");
        
        /* send error message */
        ConnectConfirmMessage err_msg(false, 0);

        // Encapsulate in TransportMessage
        TransportMessage tmsg(TransportMessage::CONNECTION, Transport::AUTHENTICATION_ERROR);
        tmsg.encapsulate(&err_msg);

        // send message
        node->sendMessage(tmsg, lnk);

        // drop link
        node->dropLink(lnk);
    }
}

// TODO (wann) wird das überhaupt aufgerufen..?  --> wie's aussieht gar nicht!
bool Transport::onLinkRequest(const ariba::NodeID& remote)
{
    if ( ! users )
        return false;
    
    
    UserPtr user = users->getUser(remote);

    if ( user )
    {
        user->set_connected(User::CONNECTION_PENDING);
        UI.write_to_user("[USER_UP] '" + user->get_name() + "' is calling ...");
        UI.write_info("'" + user->get_name() + "' is calling ...");
    }
    else
    {
        UI.write_to_user("[USER_UP] new user (" + user->get_id().getAddress() + ") is calling ...");
        UI.write_info("new user (" + user->get_id().getAddress() + ") is calling ...");
    }

    return true;
}


void Transport::onLinkDown(const ariba::LinkID& lnk, const ariba::NodeID& lscrew)
{
    if ( ! users )
        return;
    
    UserPtr user = users->getUser(lscrew);

    if ( user && (lnk == user->link || user->link == ariba::LinkID::UNSPECIFIED) )
    {
        string n = user->get_name();
        cout << n << endl;
        // clear link
        user->link = ariba::LinkID::UNSPECIFIED;

        // set disconnected
        switch ( user->get_connection_status() )
        {
            // connection errors occured  ---> set status ERROR_NOT_CONNECTED
            case User::CONNECTION_ERROR:
            {
                user->set_connected(User::ERROR_NOT_CONNECTED);
                UI.write_to_user("[USER_UP] '" + n + "' disconnected (connection errors)!");
                
                break;
            }
            
            // normal disconnect
            case User::DISCONNECT_PENDING:
            {
                user->set_connected(User::NOT_CONNECTED);
                UI.write_to_user("[USER_UP] '" + n + "' disconnected!");
                UI.write_info("User '" + n + "' disconnected!");
                
                break;
            }
            
            // connect attempt didn't work
            case User::CONNECTION_PENDING:
            {
                user->set_connected(User::NOT_CONNECTED);
                UI.write_to_user("[USER_UP] '" + n + "' disconnected!");
                UI.write_info("User '" + n + "' disconnected!");
                
                break;
            }            
            
            // connection lost ---> reconnect
            case User::CONNECTED:
            {
                UI.write_warn("Connection with '" + n + "' lost. Trying reconnect...");
                notify_conn_stat(user, false);
                
                establish_connection(user);
                
                break;
            }
            
            // ignore double LinkDown events (ariba bug here?)
            case User::NOT_CONNECTED:
            {
                break;
            }
            
            // unexpected LinkDown
            default:
            {
                ostringstream out;
                out << "Unexpected LinkDown with '" << n << "' (Connection state was: " << (int) user->get_connection_status() << ")";
                UI.write_warn(out.str());
                user->set_connected(User::NOT_CONNECTED);
                UI.write_to_user("[USER_UP]");
                
                break;
            }
        }

//        // drop user if temporal
//        if ( user->is_temporal() )
//        {
//            UI.write_to_user("DROP TEMPORAL");
//            users->dropTempUser(user);
//        }
    }
}

void Transport::onLinkChanged(const ariba::LinkID& lnk, const ariba::NodeID& lscrew)
{
    UserPtr user = users->getUser(lscrew);
    if ( user )
    {
        user->link = lnk;
        UI.write_to_user("Link changed '" + user->get_name() + "'");
    }

}


// TODO onLinkDown aufrufen?
void Transport::onLinkFail(const ariba::LinkID& lnk, const ariba::NodeID& remote)
{
    if ( ! users )
        return;

    
    UserPtr user = users->getUser(remote);
    if ( user && (lnk == user->link || user->link == ariba::LinkID::UNSPECIFIED) )
    {
        user->set_connected(User::ERROR_NOT_CONNECTED);
        UI.write_to_user("[USER_UP] '" + user->get_name() + "' disconnected!");
    }
    else
        UI.write_to_user("LINK FAIL: " + remote.getAddress());
}


/**
 * NodeListener
 */
void Transport::onJoinCompleted( const ariba::SpoVNetID& id )
{
        // Log status message
        string msg = "Joining '" + id.getAddress() + "' completed.";
        cout << msg << endl;


        // XXX testing..
        ostringstream out;
        out << msg << endl;
        for_each ( ariba::NodeID n, node->getNeighborNodes() )
        {
            out << "Neigbor: " << n.getAddress() << endl;
        }
}

void Transport::onJoinFailed( const ariba::SpoVNetID& id )
{
        // Set failure flag
        joinFailedFlag = true;

        // Log status message
        std::string msg = "Joining '" + id.getAddress() + "' failed.";
        cout << msg << endl;
}

void Transport::onLeaveCompleted( const ariba::SpoVNetID& id )
{
        // Log status message
        std::string msg = "Leaving '" + id.getAddress() + "' completed.";
        cout << msg << endl;
}

void Transport::onLeaveFailed( const ariba::SpoVNetID& id )
{
        // Log status message
        std::string msg = "Leaving '" + id.getAddress() + "' failed.";
        cout << msg << endl;
}
